package org.idea.qiyu.framework.rocketmq.config.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.idea.qiyu.framework.rocketmq.annotation.QiyuConsumer;
import org.idea.qiyu.framework.rocketmq.config.hook.MqMessageTraceHook;
import org.idea.qiyu.framework.rocketmq.config.serializer.FastJsonSerializer;
import org.idea.qiyu.framework.rocketmq.properties.RocketMqConsumerProperties;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 默认的mqconsumer模块创建工厂
 *
 * @Author linhao
 * @Date created in 5:33 下午 2021/7/2
 */
public class MqConsumerFactory implements InitializingBean,ApplicationContextAware {

    private ApplicationContext applicationContext;

    private RocketMqConsumerProperties rocketMqConsumerProperties;

    /**
     * 第一层是topic，第二层是tag
     */
    private volatile Map<String,IConcurrentlyConsumer> consumerMap = new ConcurrentHashMap<>();



    @Override
    public void afterPropertiesSet() throws Exception {
        Map<String,IConcurrentlyConsumer> consumerMap = applicationContext.getBeansOfType(IConcurrentlyConsumer.class);
        for (String consumerName : consumerMap.keySet()) {
            IConcurrentlyConsumer concurrentlyConsumer = consumerMap.get(consumerName);
            QiyuConsumer qiyuConsumer = concurrentlyConsumer.getClass().getAnnotation(QiyuConsumer.class);
            String topic = qiyuConsumer.topic();
            String tag = qiyuConsumer.tag();
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
            consumer.setVipChannelEnabled(false);
            consumer.setNamesrvAddr(rocketMqConsumerProperties.getNameSrv().trim());
            consumer.setConsumerGroup(qiyuConsumer.groupName().trim());
            consumer.setConsumeMessageBatchMaxSize(rocketMqConsumerProperties.getConsumeMessageBatchMaxSize());
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.subscribe(topic,tag);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt msg : msgs) {
                        System.out.println(msg);
                        String key = msg.getTopic() + msg.getTags();
                        IConcurrentlyConsumer iConcurrentlyConsumer = consumerMap.get(key);
                        if(iConcurrentlyConsumer!=null){
                            return iConcurrentlyConsumer.consume(msg);
                        }
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
            DefaultMQPushConsumerImpl defaultMQPushConsumer = consumer.getDefaultMQPushConsumerImpl();
            defaultMQPushConsumer.registerFilterMessageHook(new MqMessageTraceHook());
            String key = qiyuConsumer.topic() + qiyuConsumer.tag();
            consumerMap.put(key,concurrentlyConsumer);
            consumer.start();
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        this.rocketMqConsumerProperties = applicationContext.getBean(RocketMqConsumerProperties.class);
    }
}
